package cn.itcast.flink.base;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * Author itcast
 * Date 2021/7/27 17:51
 * 读取 flink-kafka topic 中的数据
 */
public class KafkaConsumerConnector {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //
        env.enableCheckpointing(6000);
        //2.Source 实例化 FlinkKafkaConsumer 配置如下参数
        Properties props = new Properties();
        /* 需要设置如下参数:
         * 1.订阅的主题
         * 2.反序列化规则
         * 3.消费者属性-集群地址
         * 4.消费者属性-消费者组id(如果不设置,会有默认的,但是默认的不方便管理)
         * 5.消费者属性-offset重置规则,如earliest/latest...
         * 6.动态分区检测(当kafka的分区数变化/增加时,Flink能够检测到!)
         * 7.如果没有设置Checkpoint,那么可以设置自动提交offset,后续学习了Checkpoint会把offset随着做Checkpoint的时候提交到Checkpoint和默认主题中*/
        //2.反序列化规则
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,SimpleStringSchema.class.getSimpleName());
        //3.消费者属性-集群地址
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");
        //4.消费者属性-消费者组id(如果不设置,会有默认的,但是默认的不方便管理)
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"flink-kafka-consumer");
        //5.消费者属性-offset重置规则,如earliest/latest...
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        //6.动态分区检测(当kafka的分区数变化/增加时,Flink能够检测到!),当前kafka分区3个，数据量上来，增大分区数，有个自动探测分区
        //机制，能够获取到当前变化的分区数，根据最新的分区个数来适配当前的Flink消费Kafka Source子任务
        //SourceFunction RichParallelSourceFunction
        props.setProperty("flink.partition-discovery.interval-millis","30000");
        //if true,将当前的 offset 提交kafka中的consumer_offset_ topic 中
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(
              "flink-kafka",
              new SimpleStringSchema(),
              props
        );
        //从最早的kafka中进行数据的消费
        consumer.setStartFromEarliest();
        //将消费的offset 提交到 checkpoint 中
        //consumer.setCommitOffsetsOnCheckpoints(true);
        DataStreamSource<String> source = env.addSource(consumer);
        //3.Sink
        source.print();
        //4.execute
        env.execute();
    }
}
